package org.research;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.apache.log4j.Logger;

public class Producer implements Runnable {

    private static final Logger log = Logger.getLogger(Producer.class);

    private Thread thisThread;
    private final ConnectionFactory connectionFactory;
    private final Connection connection;
    private final Session session;
    private final Topic topic;
    private MessageProducer producer;

    public static void start() {
        Producer producer = new Producer();
        producer.startProducingMessagesAsynchronously();
        producer.startCleaner();
    }

    private void startCleaner() {
        Runnable cleaner = new Runnable() {
            public void run() {
                try {
                    thisThread.join();
                }
                catch (InterruptedException e) {
                    log.error(e.getMessage(), e);
                }
                try {
                    connection.close();
                    log.debug("JMS connection closed.");
                }
                catch (JMSException e) {
                    log.error(e.getMessage(), e);
                }
            }
        };
        new Thread(cleaner).start();
    }

    private void startProducingMessagesAsynchronously() {
        thisThread = new Thread(this);
        thisThread.start();
    }

    public Producer() {
        try {
            InitialContext jndiContext = new InitialContext();
            connectionFactory = (ConnectionFactory) jndiContext.lookup("ConnectionFactory");
            log.debug(connectionFactory);
            connection = connectionFactory.createConnection();
            log.debug("JMS connection opened " + connection);
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            log.debug("JMS session created " + session);

            topic = (Topic) jndiContext.lookup("MyTopic");
            log.debug("Looked up topic: " + topic);

            producer = session.createProducer(topic);
        }
        catch (NamingException e) {
            throw new IllegalStateException("Unable to resolve jndi name", e);
        }
        catch (JMSException e) {
            throw new IllegalStateException("Unable to create connection", e);
        }
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                sendMessage(i);
                Thread.sleep(500);
            }
        }
        catch (JMSException e) {
            throw new IllegalStateException("Unable to send message", e);
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void sendMessage(int i) throws JMSException {
        Message message = session.createTextMessage("Message #" + i);
        message.setStringProperty("Food", "Salad");
        message.setDoubleProperty("Quantity", 12.33);
        producer.send(message);
        log.debug("Message " + i + " produced.");
    }
}
